package cn.itcast.mr.cases;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.Arrays;

/**
 * Author itcast
 * Date 2021/10/17 17:25
 * Desc TODO
 */
public class CommonFriendStepTwo {
    public static class CommonFriendsMapper extends Mapper<LongWritable, Text, Text, Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (StringUtils.isNotEmpty(line)) {
                String[] userFriendArr = line.split("\t");
                String users = userFriendArr[0];
                String friends = userFriendArr[1];
                //将其拆成两两之间关系
                String[] userList = users.split("-");
                //将用户列表排序
                Arrays.sort(userList);
                //遍历求出两两之间的关系
                for (int i = 0; i < userList.length - 1; i++) {
                    for (int j = i + 1; j < userList.length; j++) {
                        context.write(new Text(userList[i]+"-"+userList[j]),new Text(friends));
                    }
                }
            }
        }
    }
    public static class CommonFriendsReducer extends Reducer<Text, Text, Text, Text>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            StringBuffer buffer = new StringBuffer();
            for (Text friend : values) {
                buffer.append(friend);
                buffer.append("-");
            }
            buffer.deleteCharAt(buffer.length()-1);
            context.write(key,new Text(buffer.toString()));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //创建 Job 实例
        Job job = Job.getInstance();
        //job设置输入的格式化类型
        job.setInputFormatClass(TextInputFormat.class);
        //设置输入的文件路径
        TextInputFormat.addInputPath(job, new Path("D:\\project\\mr\\data\\friends_result01\\part-r-00000"));
        //job设置map的实现类
        job.setMapperClass(CommonFriendsMapper.class);
        //job设置map的输出的key类型
        job.setMapOutputKeyClass(Text.class);
        //job设置map的输出的value类型
        job.setMapOutputValueClass(Text.class);

        //job设置shuffle，设置自定义了分区
        //job设置shuffle，设置局部聚合 combiner

        //job设置reduce的实现类
        job.setReducerClass(CommonFriendsReducer.class);
        //job设置reduce的输出key
        job.setOutputKeyClass(Text.class);
        //job设置reduce的输出value
        job.setOutputValueClass(Text.class);

        //job设置reducer输出到文件的输出类型
        job.setOutputFormatClass(TextOutputFormat.class);
        //job设置输出的路径
        TextOutputFormat.setOutputPath(job, new Path("D:\\project\\mr\\data\\friends_result02"));

        //job等待执行
        boolean flag = job.waitForCompletion(true);
        //返回程序执行的结果，如果成功返回为0，否则1
        System.exit(flag ? 0 : 1);
    }
}
